package com.bs.kafka.listener;

import com.bs.kafka.service.TestService;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;

import javax.annotation.Resource;

/**
 * @author bingshao
 * @date 2023/6/9
 **/
@Configuration
@Log4j2
public class TestConsumerListener {

    @Resource
    private TestService testService;

    // kafka 消费重试有两种方式，1：使用自带的errorHandler:2：try catch自行处理

    @KafkaListener(topics = {"test-topic"},groupId = "test.consumer.group",errorHandler = "testTopicRetryHandler")
    public void consumer(ConsumerRecord<?, ?> record, Consumer consumer) {
        // TODO 幂等性先不演示跟rocketmq类似
        testService.doSomeThing();
        log.info("简单消费===Topic：" + record.topic() + "**分区" + record.partition() + "**值内容" + record.value() + "**位移" + record.offset());
        consumer.commitAsync((offsets, e) -> {
            if (e == null) {
                // 提交成功的处理逻辑
                log.info("Offset commit successful for offsets: {}", offsets);
            } else {
                // 提交失败的处理逻辑
                log.error("Offset commit failed for offsets: {}", offsets);
            }
        });
    }
}
